package com.roy.KafkaTest.stream.myexam;

import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;

public class StateStoredStreamDemo {
	public static final String APPLICATION_ID = "my-stream-app";
	public static final String BOOTSTRAP_SERVERS = "172.16.48.10:9092,172.16.48.11:9092,172.16.48.12:9092";
	public static final String TOPIC_ID = "topic2";

	public static void main(String[] args) {
		final Properties props = new Properties();
		props.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_ID);
		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
		props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
		props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
				Serdes.String().getClass());
		props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
				Serdes.String().getClass());
		props.put(StreamsConfig.STATE_DIR_CONFIG, "D:\\kfk-state-dir");

		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

		// 手动创建一个state store
		StoreBuilder<KeyValueStore<String, String>> testStateStore = Stores
				.keyValueStoreBuilder(
						// 指定state-store的名字，运行时会自动在前面配置的STATE_DIR_CONFIG路径下创建该文件夹
						Stores.persistentKeyValueStore("wyh-state-store"),
//						Stores.inMemoryKeyValueStore(name) // 内存stateStore
						Serdes.String(), Serdes.String())
				.withCachingEnabled();
		StreamsBuilder builder = new StreamsBuilder();
		// 向builder中添加state-store
		builder.addStateStore(testStateStore);
		// 创建KStreams,从wyh-topic中读入流
		KStream<String, String> inputStream = builder.stream("wyh-topic-in");
		// 使用transform实现自定义statestore,参数一是一个TransformerSupplier的实现类的对象，参数二是前面指定的state-store的名字
		KStream<String, String> transformStream = inputStream.transform(
				new TestTransformerSupplier(testStateStore.name()),
				testStateStore.name());
		// 输出topic
		transformStream.to("wyh-topic-out");
		KafkaStreams streams = new KafkaStreams(builder.build(), props);
		streams.start();
	}
}
